package xyz.hubery.service;

import com.mysql.cj.util.TimeUtil;
import io.micrometer.core.instrument.util.TimeUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.data.redis.core.ListOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import xyz.hubery.constant.Constants;
import xyz.hubery.dto.CreateKLineDto;
import xyz.hubery.enums.KlineType;
import xyz.hubery.model.Line;
import xyz.hubery.util.KlineTimeUtil;

import javax.annotation.PreDestroy;
import java.math.BigDecimal;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

@Slf4j
@Service
public class TradeKlineService implements CommandLineRunner {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    // 使用有界阻塞队列，防止内存溢出
    private static BlockingQueue<CreateKLineDto> queue = new LinkedBlockingQueue<>(1000);

    private volatile boolean running = true;

    @Override
    public void run(String... args) throws Exception {
        log.info("K线生成服务启动成功");
        // 在独立线程中处理，不阻塞主线程
        Thread processorThread = new Thread(() -> {
            while (running && !Thread.currentThread().isInterrupted()) {
                try {
                    CreateKLineDto createKLineDto = queue.poll(500, TimeUnit.MILLISECONDS);

                    if (createKLineDto != null) {
                        log.info("开始处理k线数据:{}", createKLineDto);
                        for (KlineType klineType : KlineType.values()) {
                            this.generateKLine(createKLineDto, klineType);
                        }
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    log.info("K线处理线程被中断");
                    break;
                } catch (Exception e) {
                    log.error("处理K线数据异常", e);
                }
            }
            log.info("K线处理线程已停止");
        }, "kline-processor");

        processorThread.setDaemon(true); // 设置为守护线程
        processorThread.start();
    }

    @PreDestroy
    public void destroy() {
        log.info("正在停止K线服务...");
        running = false;
    }

    /**
     * 根据当前交易生成k线
     *
     * @param klineData k线数据
     * @param klineType 当前处理的k线类型
     */
    private void generateKLine(CreateKLineDto klineData, KlineType klineType) {
        // 从redis中获取历史k线数据
        String redisKey = new StringBuilder(Constants.REDIS_KEY_TRADE_KLINE)
                .append(klineData.getSymbol().toLowerCase())
                .append(":")
                .append(klineType.getValue())  // 使用 getValue() 而不是直接 toString
                .toString();

        ListOperations<String, Object> listOperations = redisTemplate.opsForList();
        Long size = listOperations.size(redisKey);

        // 使用新的时间工具类获取时间戳
        Long curTimestamp = KlineTimeUtil.getKLineTimestamp(klineType);

        Line line = null;
        if (size == null || size == 0) { // 没有该k线历史数据
            // 使用时间戳构造 Line
            line = new Line(curTimestamp, klineData.getPrice(), klineData.getVolume());
            log.debug("创建新的K线周期: {} - {}", klineData.getSymbol(), klineType.getValue());
        } else { // 有历史数据
            // 判断是否超出容量
            if (size >= Constants.REDIS_MAX_CACHE_KLINE_SIZE) { // 移除最右侧的k线
                listOperations.rightPop(redisKey);
                log.debug("K线数据达到上限，移除最旧数据: {} - {}", klineData.getSymbol(), klineType.getValue());
            }

            // 取出当前最新k线数据
            Object recentData = listOperations.index(redisKey, 0);
            Line historyKLine;

            if (recentData instanceof Line) {
                // 如果是Line对象，直接使用
                historyKLine = (Line) recentData;
            } else if (recentData instanceof String) {
                // 如果是字符串格式，使用Line构造器解析
                historyKLine = new Line((String) recentData);
            } else {
                // 其他情况，转换为字符串再解析
                historyKLine = new Line(recentData.toString());
            }

            // 计算最高价和最低价
            BigDecimal highPrice = klineData.getPrice().compareTo(historyKLine.getHigh()) > 0 ?
                    klineData.getPrice() : historyKLine.getHigh();
            BigDecimal lowPrice = klineData.getPrice().compareTo(historyKLine.getLow()) < 0 ?
                    klineData.getPrice() : historyKLine.getLow();

            // 判断当前k线是否属于上一段k线（是否在一个时间段内）
            // 比较时间戳而不是 DateTime 对象
            if (curTimestamp.equals(historyKLine.getTimestamp())) { // 合并两个k线
                historyKLine.setClose(klineData.getPrice()); // 收盘价
                historyKLine.setHigh(highPrice); // 最高价
                historyKLine.setLow(lowPrice); // 最低价
                historyKLine.setVolume(historyKLine.getVolume().add(klineData.getVolume())); // 累加交易量

                // 更新Redis中的数据
                listOperations.set(redisKey, 0, historyKLine);
                log.debug("更新K线数据: {} - {} - 价格: {}",
                        klineData.getSymbol(), klineType.getValue(), klineData.getPrice());
                return;
            }

            // 不在一个时间段内，创建新的K线
            line = new Line(
                    curTimestamp,
                    historyKLine.getClose(), // 开盘价使用上一个周期的收盘价
                    highPrice,
                    lowPrice,
                    klineData.getPrice(),    // 收盘价使用当前价格
                    klineData.getVolume()    // 交易量
            );
            log.debug("创建新K线周期: {} - {} - 时间: {}",
                    klineData.getSymbol(), klineType.getValue(), curTimestamp);
        }

        // 添加到redis缓存
        listOperations.leftPush(redisKey, line);
        log.debug("添加新K线到Redis: {} - {} - 数据: {}",
                klineData.getSymbol(), klineType.getValue(), line);
    }

    /**
     * 添加数据到队列
     *
     * @param data
     */
    public static void addData(CreateKLineDto data) {
        boolean success = queue.offer(data);
        if (!success) {
            log.warn("K线队列已满，丢弃数据: {}", data);
        } else {
            log.debug("K线数据添加到队列: {}", data);
        }
    }

    /**
     * 获取队列当前大小（用于监控）
     */
    public static int getQueueSize() {
        return queue.size();
    }


    /**
     * 获取队列剩余容量（用于监控）
     */
    public static int getQueueRemainingCapacity() {
        return queue.remainingCapacity();
    }


}